package com.etlmaster.executor.steps.scheduler;

import java.util.concurrent.ExecutionException;

import com.etlmaster.executor.EtlMasterException;
import com.etlmaster.executor.Step;
import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.annotation.StepInstruction;
import com.etlmaster.executor.scheduler.Proc;
import com.etlmaster.executor.scheduler.ScheduleServer;
import com.etlmaster.executor.scheduler.ScheduleStatus;
import com.etlmaster.executor.scheduler.utils.CallableResult;
import com.etlmaster.executor.scheduler.utils.SubmitResult;
import com.etlmaster.executor.utils.StepState;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
@StepInstruction("submitTask")
public class SubmitTaskStep extends Step {

	private SubmitTaskStepMeta meta;

	public SubmitTaskStep(StepMeta meta) {
		super(meta);
		this.meta = (SubmitTaskStepMeta) meta;
	}

	@Override
	protected int handle() {
		String jobname = meta.getJobname();
		String taskid = meta.getTaskid();
		Boolean concurrencyRun = Boolean.parseBoolean(meta.getConcurrencyRun());
		String agentId = (String) meta.getAgentId();

		Proc proc = new Proc(jobname, taskid, concurrencyRun, meta.getParam(),
				agentId);
		proc.setParam(meta.getParam());
		meta.log("INFO", "提交任务:{}", proc.getProc_name());
		try {
			SubmitResult future = ScheduleServer.submitTask(proc);

			meta.getStepState().getReturnValue()
					.put("jobName", proc.getProc_name());
			int flag=StepState.RUNNING_SUCCESSED;
			if (future != null) {
				if (future.getResult() != null) {

					CallableResult result = (CallableResult) future.getResult()
							.get();
					
					meta.getStepState().getReturnValue()
							.put("runResult", result.getMessage());
					meta.getStepState().getReturnValue()
					.put("rtnCode", result.getResult());
			
					if(result.isError()){
						return StepState.RUNNING_FAILED;
					}
				} else if (future.getExecuteResult() == ScheduleStatus.conditionNotMeet) {
					meta.getStepState().getReturnValue()
					.put("runResult", future.getReturnMsg());
					meta.getStepState().getReturnValue()
					.put("rtnCode", ScheduleStatus.conditionNotMeet);
			
					return StepState.RUNNING_FAILED;
				}
			}else{
				meta.getStepState().getReturnValue()
				.put("runResult", ScheduleStatus.statusMap.get(ScheduleStatus.nosuchAgent));
				meta.getStepState().getReturnValue()
				.put("rtnCode", ScheduleStatus.nosuchAgent);
		
				return StepState.RUNNING_FAILED;
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
			return StepState.RUNNING_FAILED;

		} catch (ExecutionException e) {
			e.printStackTrace();
			return StepState.RUNNING_FAILED;

		} catch (EtlMasterException e) {
			e.printStackTrace();
			return StepState.RUNNING_FAILED;

		}
		return StepState.RUNNING_SUCCESSED;
	}

	@Override
	public boolean stop() {
		// TODO Auto-generated method stub
		return false;
	}

}
